package cn.huq.day03

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import java.io


object CogroupDemo {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("CogroupDemo").setMaster("local")
    val sc = new SparkContext(conf)

    val rdd1: RDD[(String, Int)] = sc.parallelize(List(("tom", 1), ("kitty", 2), ("tom", 2), ("jetty",3)), 2)
    val rdd2: RDD[(String, Int)] = sc.parallelize(List(("jetty", 1), ("shuke", 2), ("tom", 1)), 2)

//    val grouped: RDD[(String, Iterable[Int])] = rdd1.groupByKey()
    // cogroup是将两个或多个rdd，合并到一起进行分组【与group区别是，group只能是对一个rdd进行分组】
    val cogrouped: RDD[(String, (Iterable[Int], Iterable[Int]))] = rdd1.cogroup(rdd2)

//    println(cogrouped.collect().toBuffer)
    //使用cogroup实现wordcount功能，不推荐，效率低！【建议：使用union将多个RDD聚合在一起，然后使用reduceByKey】
    val wcRDD: RDD[(String, Int)] = cogrouped.mapValues(t => (t._1.sum + t._2.sum))
//    println(wcRDD.collect().toBuffer)

    // cogroup要求key的类型一定相同，但是value的类型可以不相同
    // 一个rdd最多可以cogroup 3个rdd
    val rdd3: RDD[(String, Double)] = sc.parallelize(List(("tom", 2.0)), 2)
    val cogroup3: RDD[(String, (Iterable[Int], Iterable[Int], Iterable[Double]))] = rdd1.cogroup(rdd2, rdd3)

    sc.stop()
  }

}
